/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.core.cluster;

import java.io.FileOutputStream;
import java.net.InetAddress;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.NoSerializable;
import org.apache.mina.common.SessionClosedListener;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPTaskException;
import com.cqcis.uip.base.framework.base.impl.DefaultTelnetConnector;
import com.cqcis.uip.base.framework.codec.text.HttpProtocolDecoder;
import com.cqcis.uip.base.framework.codec.text.HttpProtocolEncoder;
import com.cqcis.uip.base.framework.codec.text.TextProtocolDecoder;
import com.cqcis.uip.base.framework.codec.text.TextProtocolEncoder;
import com.cqcis.uip.base.framework.common.ControlMsg;
import com.cqcis.uip.base.framework.common.HttpParam;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.common.Task.Source;
import com.cqcis.uip.base.framework.common.Task.StateChangeListener;
import com.cqcis.uip.base.framework.common.Task.Type;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.Service;
import com.cqcis.uip.base.framework.core.impl.TelnetConnectorServiceAdaptor;
import com.cqcis.uip.base.framework.core.service.RegisteredServiceContextImpl;
import com.cqcis.uip.base.framework.handlers.ClientService;
import com.cqcis.uip.base.framework.handlers.CommonService;
import com.cqcis.uip.base.framework.handlers.TelnetClientService;
import com.cqcis.uip.base.framework.handlers.impl.DefaultClientHandler;
import com.cqcis.uip.base.framework.handlers.impl.DefaultTelnetClientAdaptor;
import com.cqcis.uip.base.framework.utils.ByteBuffer;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.EndpointGenerator;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.LoopThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.SessionUtil;
import com.cqcis.uip.base.framework.utils.StringUtil;

/**
 * 集群终端选择, cluster必须互相配信任
 * 
 * @author huangmy
 * @date   2009-1-3
 */
public class ClusterSupportSelector implements CommonService, Service, ActiveClusterService, NoSerializable {
	private static final AtomicLong idNext = new AtomicLong();
	private static Log log = LogFactory.getLog(ClusterSupportSelector.class);
	private static String CT_COUNTDOWN = "CT_COUNTDOWN_" + ClusterSupportSelector.class.getName();
	private static String CT_DEALING = "CT_DEALING_" + ClusterSupportSelector.class.getName();
	private static String WAITE_NOTE = "WAITE_NOTE_" + ClusterSupportSelector.class.getName();
	private static String TELNET_PROMPT = "uip>"; // prompt
	private static String LOGIN_PROMPT = "login:";
	private static String PASSWORD_PROMPT = "Password:";
    private Map<String, TelnetSwitchHttpProctolStub> stubs = new ConcurrentHashMap<String, TelnetSwitchHttpProctolStub>();
	private RegisteredServiceContextImpl serviceContext; // context
	
	public ClusterSupportSelector() {		
	}
	
	/**
	 * 获取服务单例
	 * 
	 * @param serviceContext
	 * @return
	 */
	public final static ClusterSupportSelector get(RegisteredServiceContext serviceContext) {
		ServiceReference sr = serviceContext.getBundleContext().getServiceReference(ClusterSupportSelector.class.getName());
		if (sr != null) {
			return (ClusterSupportSelector)serviceContext.getBundleContext().getService(sr);
		}
		
		return null;
	}
	
	/**
	 * 获取服务单例
	 * 
	 * @param serviceContext
	 * @return
	 */
	public final static ClusterSupportSelector get(BundleContext bundleContext) {
		ServiceReference sr = bundleContext.getServiceReference(ClusterSupportSelector.class.getName());
		if (sr != null) {
			return (ClusterSupportSelector)bundleContext.getService(sr);
		}
		
		return null;
	}
	
	/**
	 * 获取关系
	 * 
	 * @return
	 */
	private ClusterRelationService getClusterRelationService() {
		BundleContext bundleContext = serviceContext.getBundleContext();
		ServiceReference sr = bundleContext.getServiceReference(ClusterRelationService.class.getName());
		if (sr == null) {
			return null;
		}
		
		return (ClusterRelationService)bundleContext.getService(sr);
	}
	
	private void publishActiveClusterService() {
		BundleContext bundleContext = serviceContext.getBundleContext();
		bundleContext.registerService(ActiveClusterService.class.getName(), this, null);
	}
		
	public void start(RegisteredServiceContext serviceContext) throws Exception {
		this.serviceContext = (RegisteredServiceContextImpl)serviceContext;
		// publish
		publishActiveClusterService();
	}
	
	public void stop(RegisteredServiceContext serviceContext) throws Exception {	
	    if (this.stubs != null) {
	    	for (TelnetSwitchHttpProctolStub stub : this.stubs.values()) {
	    		stub.stop(stub.getRegisteredServiceContext());
	    	}
	    }
	}
	
	public Collection<ClusterStub> getStubs() {	
		List<ClusterStub> stubs = new ArrayList<ClusterStub>();
		Collection<TelnetSwitchHttpProctolStub> protocolStubs = this.stubs.values();
		for (TelnetSwitchHttpProctolStub protocolStub : protocolStubs) {
			stubs.add(new ClusterStub(protocolStub));
		}
		
		return stubs;
	}
	
	public void remove(String constrainKey) {
		TelnetSwitchHttpProctolStub stub = this.stubs.remove(constrainKey);
		if (stub != null) {
			try {
				stub.stop(this.serviceContext);
			} catch (Exception e) {
                throw new UIPTaskException(e);
			}
		}
	}
	
	public void active() {
		ClusterRelationService relationService = getClusterRelationService();
		if (relationService == null) {
			return;
		}
		
		ClusterRelation relation = null;
		
		try {
			relation = relationService.getRelations();
		} catch (Exception e) {
			return;
		}
		
		// 1. check
		Map<String, String> relas = relation.getRelations();
		for (Entry<String, String> entry : relas.entrySet()) {
			String nodeZName = entry.getKey();
			String nodeAName = entry.getValue();
			
			ClusterNode nodeZ = relation.getNodes().get(nodeZName);
			ClusterNode nodeA = relation.getNodes().get(nodeAName);
			
			if (nodeZ == null) {
				throw new UIPTaskException("Can't find NODE for name: " + nodeZName);
			}
			if (nodeA == null) {
				throw new UIPTaskException("Can't find NODE for name: " + nodeAName);
			}
		}
		
		// 2. active
		for (Entry<String, String> entry : relas.entrySet()) {
			String nodeZName = entry.getKey();
			String nodeAName = entry.getValue();
									
			activeStub(nodeZName, nodeAName);
		}
	}
	
	private void activeStub(String nodeZName, String nodeAName) {
		Task task = new Task(FrameworkConstants.CLUSTER_CT);
		Map<String, String> props = new HashMap<String, String>();
		props.put(EndpointGenerator.ENDPOINT_EXPRESSION_LOCATION, nodeZName);
		task.route().to(EndpointGenerator.generate(Endpoint.ACTIVE_ID, props));
		
		Map<String, String> propsFrom = new HashMap<String, String>();
		propsFrom.put(EndpointGenerator.ENDPOINT_EXPRESSION_LOCATION, nodeAName);
		task.route().from(EndpointGenerator.generate(Endpoint.ACTIVE_ID, propsFrom));
		
		task.setType(Type.ACTIVE);
		
		try {
			this.schedule(task);
		} catch (Exception e) {
			log.error("Schedule for active Cluster Error, nodeZName: " + nodeZName
					+ ", nodeAName: " + nodeAName, e);
		}
	}
	
	public void active(long id) {
		TelnetSwitchHttpProctolStub findStub = null;
		for (TelnetSwitchHttpProctolStub stub : this.stubs.values()) {
			if (stub.id == id) {
				findStub = stub;
				break;
			}
		}
		
		if (findStub == null) {
			return;
		}
		
		activeStub(findStub.si.nodeName, findStub.si.nodeNameA);
	}
	
	/**
	 * 以session为模板，创建或获取给定的stub
	 * 
	 * @param sesson
	 * @return
	 */
	public TelnetSwitchHttpProctolStub createOrGetStub(IoSession session, SwitchInfo si) throws Exception {
		TelnetSwitchHttpProctolStub stub = new TelnetSwitchHttpProctolStub(si) {
			@Override
			protected DefaultTelnetClientAdaptor createTelnetClient() {
				return createFlipTelnetClientAdaptor(this);
			}
		};
		
		if (log.isDebugEnabled()) {
			log.debug("Create model Stub for session: " + session + ", SI: " + si);
		}
		
		this.stubs.put(si.constrainKey, stub);
		
		// 1. init
		stub.init(this.serviceContext); 
		
		// 2. start now
		RegisteredServiceContextImpl serviceContext = this.serviceContext.copyFrom(stub, null);
		stub.start(serviceContext);
		
		// session injected
		stub.clientNew(session);
						
		return stub;
	}
	
	/**
	 * 激活
	 * 
	 * @param task
	 * @param stub
	 * @return
	 */
	private Task createActiveFrom(Task task, final TelnetSwitchHttpProctolStub stub) {
		// 执行一次cluster连接
		Task ct = new Task(FrameworkConstants.CLUSTER_CT);
		ct.setSource(Source.REMOTE);
		ct.setType(Type.ACTIVE);
		ct.setAttachement(null);
		ct.route().from(task.route().from());
		ct.route().to(task.route().to());
		ct.addStateChangeListener(new StateChangeListener() {
			private static final long serialVersionUID = 1L;
			public void changed(Task t, State state) {
				if (state == State.ERROR) {
					stub.state = StubState.DEACTIVE;
					stub.stateNode = t.getMessage();
					IoSession session = t.getSession();
					if (session != null) {
						session.close();
					}
				}
			}
		});
		
		return ct;
	}
    
	public void schedule(Task task) throws Exception {
		if (log.isDebugEnabled()) {
			log.debug("schedule task: \n" + task);
		}
		
		TelnetSwitchHttpProctolStub stub = getStub(task);
		if (stub == null) {
			throw new UIPTaskException("Can't find stub for this task: " + task);
		}
		
		// SET task'route from and to
		SwitchInfo si = stub.si;
		task.route().to().setAttribute(Endpoint.LOCATION, si.toString());		
		task.setSource(Source.REMOTE);	
		
		try {
			boolean isSchedule = false;
			Task ct = null;
			
			if (stub.state == StubState.DEACTIVE) {
				ct = createActiveFrom(task, stub);
				
				isSchedule = true;
			} else if (task.getType() == Type.ACTIVE
					&& stub.state == StubState.DEACTIVE) {
				isSchedule = true;
				ct = task;
			}
			
			if (isSchedule) {
				if (!stub.schedule(ct)) {
					throw new UIPTaskException(ct.getMessage() + ", Can't create cluster for: " + task.route(), task.getThrowable());
				}
			}
		} catch (Exception et) {
			if (stub.stateNode == null) {
				stub.stateNode = et.getMessage();
			}
			throw et;
		}
		
		if (!FrameworkConstants.CLUSTER_CT.equalsIgnoreCase(task.getTaskId())
				&& task.getType() != Type.ACTIVE) {
			stub.taskQueue.offer(task);
		}		
	}
	
	private static String getHostAddress() throws UnknownHostException {
		InetAddress host = InetAddress.getLocalHost();
	    return host.getHostAddress();
	}
	
	/**
	 * 处理发送数据
	 * 
	 * @param source
	 * @param session
	 * @throws Exception 
	 */
	private void forSendCluster(SwitchInfo si, 
			ConcurrentTaskQueue<Task> source, IoSession session) {		
		Task task = null;
		while ((task = source.poll()) != null) {
			try {
			    forSendSingleton(task, null, si, session);
			} catch (Throwable t) {
				log.error("Catch Exception in cluster Send.", t);
				// Exception
				task.setState(State.ERROR);
				task.setMessage(t.getMessage(), t);
				
				if (Type.NORMAL == task.getType()) {
					// result
					Endpoint fromEndpoint = task.route().from();	
					fromEndpoint.removeAttribute(EndpointGenerator.ENDPOINT_EXPRESSION_LOCATION);
					
					RegisteredService rs = getRS(fromEndpoint); // 任务源服务
					if (rs != null) {					
						rs.processResult(task);
					}
				}
			}
		}
	}
	
	private static String getHost(String host) {
		host = StringUtil.trimWhiteToNull(host);		
		if ("127.0.0.1".equalsIgnoreCase(host)
				|| "localhost".equalsIgnoreCase(host)) {
			try {
				host = getHostAddress();
			} catch (UnknownHostException e) {
                throw new UIPTaskException(e);
			}
		}
		
		return host;
	}
	
	/**
	 * SEND
	 * 
	 * @param task
	 * @param si
	 * @param session
	 */
	private HttpParam forSendSingleton(Task task, Type specialType, SwitchInfo si, IoSession session) throws Exception {
		HttpParam param = null;
		Endpoint toEndpoint = task.route().to();
				
		SwitchInfo sito = parseFromConstrainKey(toEndpoint.getAttribute(
				Endpoint.LOCATION));
		sito.address = getHost(sito.address);
		sito.addressA = getHost(sito.addressA);
				
		toEndpoint.setAttribute(Endpoint.LOCATION, sito.toString());
							
		param = ClusterSwitchSupport.getPostParams(task, session);
		param.setHeaderline(getHttpPostHeaderLine(si));
		param.addParams("Host", si.address + ":" + si.port);
		param.addParams(ClusterSwitchSupport.PARAM_TASK_TYPE, 
				specialType != null ? specialType.getName() : task.getType().getName());
						
		if (log.isDebugEnabled()) {
			log.debug("Send Param: \n" + param.getHeadersString() + ", session: " + session);
		}
		
		if (session != null) {
			session.write(param);
		}
		
		return param;
	}
	
	/**
	 * NODE
	 * 
	 * @param nodeName
	 * @return
	 */
	private SwitchInfo getInfoFromClusterRelation(String nodeName) throws Exception {
		ClusterRelationService clusterService = getClusterRelationService();
		if (clusterService == null) {
			throw new UIPTaskException("Cluster Config Service is NOT Active.");
		}
		
		ClusterRelation relation = clusterService.getRelations();
		if (relation == null) {
			throw new UIPTaskException("Cluster Config is Empty.");
		}
		
		ClusterNode node = relation.getNodes().get(nodeName);
		if (node == null) {
			throw new UIPTaskException("Can't find NODE config for nodeName: " + nodeName);
		}
		
		String nodeAName = relation.getRelations().get(nodeName);
		if (!StringUtil.hasText(nodeAName)) {
			throw new UIPTaskException("Can't find Rela NODE for nodeName: " + nodeName);
		}
		
		ClusterNode nodeA = relation.getNodes().get(nodeAName);
		if (nodeA == null) {
			throw new UIPTaskException("Can't find NODE config for nodeName: " + nodeAName);
		}
		
		// SI
		SwitchInfo si = new SwitchInfo();
		si.username = node.getUsername();
		si.password = node.getPassword();
		si.address = getHost(node.getHost());
		si.port = node.getPort();
		si.path = "/";
		si.usernameA = nodeA.getUsername();
		si.passwordA = nodeA.getPassword();
		si.addressA = getHost(nodeA.getHost());
		si.portA = nodeA.getPort();
		si.nodeName = nodeName;
		si.nodeNameA = nodeA.getName();
		si.constrainKey = getSIConstrainKey(si);
		
		return si;
	}
	
	/**
	 * 获取STUB
	 * 
	 * @param task
	 * @return
	 * @throws Exception
	 */
	private TelnetSwitchHttpProctolStub getStub(Task task) throws Exception {				
		SwitchInfo si = getSwitchStubInfo(task);
		if (si != null) {
			si = getInfoFromClusterRelation(si.nodeName);
		} else {
			throw new UIPTaskException("Task Source Error for 'location' attribute.");
		}
		
		TelnetSwitchHttpProctolStub stub = this.stubs.get(si.constrainKey);
		if (log.isDebugEnabled()) {
			log.debug("Get Stub for ConstrainKey: " + si.constrainKey + 
					", STUB: " + stub);
		}
		
		if (stub == null) {			
			stub = new TelnetSwitchHttpProctolStub(si);
			this.stubs.put(si.constrainKey, stub);
			
			// 1. init
			stub.init(this.serviceContext); 
			
			// 2. start now
			RegisteredServiceContextImpl serviceContext = this.serviceContext.copyFrom(stub, null);
			stub.start(serviceContext);
			
			return stub;
		}
				
		return stub;
	}
	
	/**
	 *  反向 service
	 * 
	 * @param stub
	 * @return
	 */
	private ClientService getFlipClientService(final TelnetSwitchHttpProctolStub stub) {
		return new ClientService() {
			public Object doMessageSend(final Task task) throws Exception {
				HttpParam param = forSendSingleton(task, Type.ACTIVE, stub.si, null);
				
				IoSession session = task.getSession();
				param.setSession(session);
				
				session.setAttribute(FrameworkConstants.CLUSTER_SWITCH);
				
				stub.clusterWaitTpe.execute(new Runnable() {
					public void run() {
						try {
							forProcessTaskSend(stub, null, task.getSession());
						} catch (Exception e) {
							stub.getRegisteredServiceContext().catchException(task, e);
						}
					};
				});
				
				if (log.isDebugEnabled()) {
					log.debug("Send flip Param: " + param.getHeadersString());
				}
			    
				return param;
			}
			
			public void doMessageReceived(Object message, Task task)
					throws Exception {
				HttpParam param = (HttpParam)message;
				IoSession session = task.getSession();
				
				task.setExecuteDefault(false); 
				task.setGenerateResult(false);
				
				doForTaskReceived(param, session, stub, true); // processor				
			}
		};
	}
	
	/**
	 * 正常client STUB
	 * 
	 * @return
	 */
	private TelnetClientService getNormalTelnetClientService(final TelnetSwitchHttpProctolStub stub) {
		return new TelnetClientService() {
			public String doTextlineSend(Task task) throws Exception {						
				return FrameworkConstants.CLUSTER_SWITCH;
			}
			
			public void doTextlineReceived(String textline, final Task task)
					throws Exception {
				task.setExecuteDefault(false); 
				task.setGenerateResult(false);
				
				if (textline.indexOf(TELNET_PROMPT) >= 0) {
					// cluster success.
					final IoSession session = task.getSession();
					session.setAttribute(FrameworkConstants.CLUSTER_SWITCH);
															
					final CountDownLatch ctDown = new CountDownLatch(1);
					session.setAttribute(CT_COUNTDOWN, ctDown);
					session.setAttribute(CT_DEALING, Boolean.TRUE);
					
					// for CT
					forSendSingleton(task, Type.ACTIVE, stub.si, session);	
					
					stub.clusterWaitTpe.execute(new Runnable() {
						public void run() {
							try {
								forProcessTaskSend(stub, ctDown, session);
							} catch (Exception e) {
								stub.getRegisteredServiceContext().catchException(task, e);
							}
						};
					});
				}
			}
		};
	}
	
	/**
	 * normal client
	 * 
	 * @return
	 */
	private ClientService getNormalClientService() {
		return new ClientService() {
			public Object doMessageSend(Task task) throws Exception {
				// do nothing
				return null;
			}
			
			public void doMessageReceived(Object message, Task task)
					throws Exception {
				HttpParam param = (HttpParam)message;
				IoSession session = task.getSession();
				
				task.setExecuteDefault(false); 
				task.setGenerateResult(false);
				
				doForTaskReceived(param, session, null, false); // processor
			}
		};
	}
	
	/**
	 * 设置
	 * 
	 * @param tca
	 * @param stub
	 */
	private void setServiceForClient(DefaultTelnetClientAdaptor tca,
			         TelnetSwitchHttpProctolStub stub) {
		// telnet service
		tca.setTelnetClientService(getNormalTelnetClientService(stub));
		// object service
		tca.setSocketClientService(getNormalClientService());
	}
	
	/**
	 * normal client
	 * 
	 * @param stub
	 * @return
	 */
	private DefaultTelnetClientAdaptor createProtocolTelnetClientAdaptor(TelnetSwitchHttpProctolStub stub) {
		DefaultTelnetClientAdaptor tca = new DefaultTelnetClientAdaptor();
		// service
		setServiceForClient(tca, stub);
		
		return tca;
	}
	
	/**
	 * flip client
	 * 
	 * @param stub
	 * @return
	 */
	private DefaultTelnetClientAdaptor createFlipTelnetClientAdaptor(TelnetSwitchHttpProctolStub stub) {
		DefaultTelnetClientAdaptor tca = new DefaultTelnetClientAdaptor() {
			@Override
			protected int getLoggingStateFrom(Task task) {
				return TelnetControlMsg.LOGIN;
			}
		};
		// object service
		tca.setSocketClientService(getFlipClientService(stub));
		
		return tca;
	}
	
	/**
	 * 处理 发送
	 * 
	 * @param stub
	 * @param ctDown
	 * @param session
	 * @throws Exception
	 */
	private void forProcessTaskSend(final TelnetSwitchHttpProctolStub stub, final CountDownLatch ctDown, 
			final IoSession session) throws Exception {
		session.addSessionClosedListener(new SessionClosedListener() {
			public void sessionClosed(IoSession session) {
				stub.state = StubState.DEACTIVE;
				stub.stateNode = "Cluster Closed.";
				
				Task t = null;
				while ((t = stub.taskQueue.remove()) != null) {
					t.setState(State.ERROR);
					t.setMessage("Cluster Closed.");
					
					RegisteredService rs = getRS(t.route().from());
					if (rs != null) {
						rs.processResult(t);
					}
				}
				
				notifyClusterFinish(session);
			}
		});
		
		if (ctDown != null) {
			ctDown.await(60, TimeUnit.SECONDS); // WAIT
			if (session.getAttribute(CT_DEALING) != null) {
				stub.state = StubState.DEACTIVE;
				String note = (String)session.removeAttribute(WAITE_NOTE);
				stub.stateNode = note != null ? note : "Cluster failure for timeout 60s.";
				throw new UIPTaskException(stub.stateNode);
			}
			
			stub.state = StubState.ACTIVE;
			stub.stateNode = "initiative";
		}
		
		stub.clusterSendTpe.execute(new Runnable() {
			public void run() {							
				forSendCluster(stub.si, stub.taskQueue, session);
			}
		});	
	}
	
	private void notifyClusterFinish(IoSession session) {
		CountDownLatch ctDown = (CountDownLatch)session.removeAttribute(CT_COUNTDOWN);
    	if (ctDown != null) {
    		ctDown.countDown();
    	}
	}
	
	/**
	 * 处理接收
	 * 
	 * @param param
	 * @param session
	 * @throws Exception
	 */
	private void doForTaskReceived(HttpParam param, IoSession session, TelnetSwitchHttpProctolStub stub
			 , boolean flip) throws Exception {
		Type type = ClusterSwitchSupport.getType(param);
		Endpoint to = getToEndpoint(param); // to endpoint
		
		if (log.isDebugEnabled()) {
			log.debug("Recv Param: \n" + param.getHeadersString() +
					"\n, on session: " + session);
		}
								
		if (Type.ACTIVE == type) { // for active
			Task task = (Task)param.getContents().getObject(Task.class.getClassLoader());				
			if (log.isDebugEnabled()) {
				log.debug("TaskReceived Active Task: \n" + task);
			}
			
			if (param.isOK() || task.getState() != State.ERROR) {
				// CT success
		    	task.setType(Type.RESULT);
				task.setState(State.SUCCESS);
				task.setMessage(Type.ACTIVE.getName());
								
				HttpParam resultP = ClusterSwitchSupport.getResponseParams(task, param);
				
				if (log.isDebugEnabled()) {
					log.debug("TaskReceived Result Param: \n" + resultP.getHeadersString());
				}
				
				session.write(resultP);// result
				SessionUtil.setTimeoutCheckClose(session, true);
				
				session.removeAttribute(CT_DEALING);
			} else {
				session.setAttribute(WAITE_NOTE, param.getHeaderline());
			}

			notifyClusterFinish(session);				    	
			return;
		}
		
		if (Type.RESULT == type && to.getId() == Endpoint.ACTIVE_ID) {
			if (!param.isOK()) {
				session.setAttribute(WAITE_NOTE, param.getHeaderline());
				notifyClusterFinish(session);
			} else if (flip) {
				stub.state = StubState.ACTIVE;
				stub.stateNode = "passive";
				SessionUtil.setTimeoutCheckClose(session, true);
			}
			return;
		}
						
		try {				    
		    RegisteredService rs = getRS(to);				    
		    if (!rs.scheduleCluster(param)) {
		    	throw new UIPTaskException("scheduleCluster return false, " +
		    			"for the Endpoint: " + to);
		    }
		} catch(Throwable t) {
			log.error("Catch Exception in scheduleCluster", t);
			// result
			type = ClusterSwitchSupport.getType(param);
			if (Type.NORMAL == type || Type.ACTIVE == type) {
				param.setHeaderline("HTTP/1.1 500 Internal Server Error");
				session.write(param);
			}
		}
	}
	
	private Endpoint getToEndpoint(HttpParam param) throws Exception {
		Endpoint endpoint = ClusterSwitchSupport.getToEndpoint(param);
	    if (endpoint == null) {
	    	throw new UIPTaskException("Can't find the endpoint for " +
	    			"the Param: " + param);
	    }
	    
	    return endpoint;
	}
	
	private RegisteredService getRS(Endpoint to) {
		RegisteredService rs = serviceContext.getServiceRepository()
				.getRegisteredService(to);
		if (rs == null) {
			throw new UIPTaskException("Can't find the RS for the "
					+ "Endpoint: " + to);
		}
		
		return rs;
	}
	
	/**
	 * POST line
	 * 
	 * @param si
	 * @return
	 */
	private String getHttpPostHeaderLine(SwitchInfo si) {
		String line = "POST " + si.path + " HTTP/1.1";
		return line;
	}
	
	/**
	 * 获取远程连接信息
	 * 
	 * @param task
	 * @return
	 * @throws URISyntaxException
	 */
	private SwitchInfo getSwitchStubInfo(Task task) throws URISyntaxException {
		Endpoint endpoint = task.route().to();
		String nodeName = endpoint.getAttribute(Endpoint.LOCATION);
		if (StringUtil.hasText(nodeName)) {
			SwitchInfo si = new SwitchInfo();
			si.nodeName = nodeName;
			return si;
		}
				
		return null;
	}
	
	private static String getSIConstrainKey(SwitchInfo si) {
		StringBuffer key = new StringBuffer();
		key.append(si.nodeName);
		key.append("(");
		key.append(si.username);
		key.append(":");
		key.append(si.password);
		key.append("@");
		key.append(si.address);
		key.append(":");
		key.append(si.port);
		key.append(")<-");
		key.append(si.nodeNameA);
		key.append("(");
		key.append(si.usernameA);
		key.append(":");
		key.append(si.passwordA);
		key.append("@");
		key.append(si.addressA);
		key.append(":");
		key.append(si.portA);
		key.append(")");
		
		return key.toString();
	}
	
	/**
	 * 解析
	 * 
	 * @param key
	 * @return
	 */
	public static SwitchInfo parseFromConstrainKey(String key) {
		SwitchInfo si = new SwitchInfo();
		int split = key.indexOf("(");
		int splitPre = split;			
		si.nodeName = key.substring(0, split);
		splitPre = split;
		split = key.indexOf(":",splitPre + 1);		
		si.username = key.substring(splitPre + 1, split);	
		splitPre = split;
		split = key.indexOf("@", splitPre + 1);
		si.password = key.substring(splitPre + 1, split);
		splitPre = split;
		split = key.indexOf(":",splitPre + 1);
		si.address = getHost(key.substring(splitPre + 1, split));
		splitPre = split;
		split = key.indexOf(")<-",splitPre + 1);
		si.port = Integer.valueOf(key.substring(splitPre + 1, split));
		si.path = "/";
		splitPre = split + 1 + 1;
		split = key.indexOf("(",splitPre + 1);
		si.nodeNameA = key.substring(splitPre + 1, split);
		splitPre = split;
		split = key.indexOf(":",splitPre + 1);
		si.usernameA = key.substring(splitPre + 1, split);
		splitPre = split;
		split = key.indexOf("@",splitPre + 1);		
		si.passwordA = key.substring(splitPre + 1, split);
		splitPre = split;
		split = key.indexOf(":",splitPre + 1);
		si.addressA = getHost(key.substring(splitPre + 1, split));
		splitPre = split;
		split = key.indexOf(")",splitPre + 1);
		si.portA = Integer.valueOf(key.substring(splitPre + 1, split));
		si.constrainKey = key;
		
		return si;
	}
		
	/**
	 * 连接相关信息
	 * 
	 * @author huangmy
	 * @date   2009-1-3
	 */
	public static class SwitchInfo {
		private String constrainKey;		
		private String username;
		private String password;
		private String address;
		private int port;
		private String path = "/";
		
		private String nodeName;// NODE_NAME
		
		private String usernameA;
		private String passwordA;
		private String addressA;
		private int portA;
		private String nodeNameA;
		
		/**
		 * 反转
		 * 
		 * @return
		 */
		public SwitchInfo flip() {
			SwitchInfo si = new SwitchInfo();
			si.nodeName = this.nodeNameA;
			si.username = this.usernameA;
			si.password = this.passwordA;
			si.address = this.addressA;
			si.port = portA;
			si.path = this.path;
			si.nodeNameA = this.nodeName;
			si.usernameA = this.username;
			si.passwordA = this.password;
			si.addressA = this.address;
			si.portA = this.port;
			si.constrainKey = getSIConstrainKey(si);
			
			return si;
		}
		
		public String getDesc() {
			return nodeName + "(" + nodeNameA + ")";
		}
		
		@Override
		public String toString() {
			return getSIConstrainKey(this);
		}

		public String getConstrainKey() {
			return constrainKey;
		}

		public String getUsername() {
			return username;
		}

		public String getPassword() {
			return password;
		}

		public String getAddress() {
			return address;
		}

		public int getPort() {
			return port;
		}

		public String getPath() {
			return path;
		}

		public String getNodeName() {
			return nodeName;
		}

		public String getUsernameA() {
			return usernameA;
		}

		public String getPasswordA() {
			return passwordA;
		}

		public String getAddressA() {
			return addressA;
		}

		public int getPortA() {
			return portA;
		}

		public String getNodeNameA() {
			return nodeNameA;
		}
	}
	
	/**
	 * Get the HANDLER
	 * 
	 * @return
	 */
	private DefaultTelnetClientAdaptor getClientAdaptor(SwitchInfo si, TelnetSwitchHttpProctolStub stub) {
		DefaultTelnetClientAdaptor tca = createProtocolTelnetClientAdaptor(stub);
		tca.setLogincheck(true);
		tca.setExitAfterCmdExecuted(false);
		tca.setPrompt(TELNET_PROMPT);
		tca.setUserName(si.username);
		tca.setPassword(si.password);
		tca.setLoginPrompt("login:");
		tca.setPasswordPrompt("Password:");
		tca.setLoginTimeout(60000);
		
		return tca;
	}

	/**
	 * 协议切换解码
	 * 
	 * @return
	 */
	private TextProtocolDecoder getTextProtocolDecoder() {
		final HttpProtocolDecoder hd = new HttpProtocolDecoder();
		TextProtocolDecoder td = new TextProtocolDecoder() {
			@Override
			public Object decode(ByteBuffer buffer) throws Exception {
				IoSession session = buffer.getSession();
				Object sw = session.getAttribute(FrameworkConstants.CLUSTER_SWITCH);
				if (sw != null) {
					HttpParam p = (HttpParam)hd.decode(buffer);
					if (p == null) {
						return null;
					}
															
					ControlMsg msg = new ControlMsg();
					msg.setAttribute(FrameworkConstants.CLUSTER_SWITCH, p);
					return msg;
				}
				
				return super.decode(buffer);
			}
		};
		td.setCharset("GBK");
		td.setPrintLog(false);
		
		List<String> prompts = new ArrayList<String>();
	    prompts.add(LOGIN_PROMPT);
	    prompts.add(PASSWORD_PROMPT);
		prompts.add(TELNET_PROMPT);
		td.setLinePrompt(prompts);
		
		return td;
	}
	
	/**
	 * 协议切换编码
	 * 
	 * @return
	 */
	private TextProtocolEncoder getTextProtocolEncoder() {
		final HttpProtocolEncoder he = new HttpProtocolEncoder();
		TextProtocolEncoder te = new TextProtocolEncoder() {
		    int count = 0;
			@Override
			public ByteBuffer encode(Object message) throws Exception {
				if (message instanceof HttpParam) {
					ByteBuffer buf = he.encode(message);
					
					int oldpos = buf.position();
					FileOutputStream is = new FileOutputStream("D:/file/" + (++ count) + ".bak");
					is.getChannel().write(buf.buf());
					
					buf.position(oldpos);
					
					is.close();
					
					return buf;
				}
				return super.encode(message);
			}
		};
		te.setCharset("GBK");
		te.setPrintLog(false);
		
		return te;
	}
	
	public static enum StubState {
		ACTIVE("ACTIVE"),
		DEACTIVE("DEACTIVE");		
		
		String name;
		
		StubState(String name) {
			this.name = name;
		}
		
		@Override
		public String toString() {
			return this.name;
		}
		
		String getName() {
			return this.name;
		}
	}
	
	/**
	 * STUB
	 * 
	 * @author huangmy
	 * @date   2009-1-3
	 */
    class TelnetSwitchHttpProctolStub extends DefaultTelnetConnector {   
		private static final long serialVersionUID = 1L;
		private ConcurrentTaskQueue<Task> taskQueue = new ConcurrentTaskQueue<Task>();
    	protected SwitchInfo si; // info
    	protected StubState state = StubState.DEACTIVE; // 是否是新建， 对于新建stub, 会先执行登录和集群建立操作
    	protected String stateNode = "Clustering Wait";
    	private LoopThreadPoolExecutor clusterSendTpe;
    	private ThreadPoolExecutor clusterWaitTpe;
    	protected long id;
    	
    	public TelnetSwitchHttpProctolStub(SwitchInfo si) {
    		this.si = si;
    		this.id = idNext.incrementAndGet();
    	}
    	
    	@Override
    	protected DefaultClientHandler getClientHandler() {
    		DefaultClientHandler handler = new DefaultClientHandler() {
    			@Override
    			protected void doSessionClosed(IoSession session)
    					throws Exception {    				
    				super.doSessionClosed(session);
    				TelnetSwitchHttpProctolStub stub = stubs.get(si.constrainKey);
    				if (stub != null) {
    					stub.state = StubState.DEACTIVE;
    				}
    			}
    		};
    		return handler;
    	}
    	
    	@Override
    	protected long getBundleId(RegisteredServiceContext serviceContext) {
    		return serviceContext.getBundleContext().getBundle().getBundleId();
    	}
    	
    	@Override
    	public void start(RegisteredServiceContext serviceContext)
    			throws Exception {    		
    		super.start(serviceContext);
    		
    		this.clusterSendTpe = LoopThreadPoolExecutor.getExecutor(1, 2, 60, 
    				serviceContext, "Cluster " + getInterfaceName(), 10000);
    		this.clusterWaitTpe = LogServiceThreadPoolExecutor.getLogExecutor(1, 2, 60, 
    				serviceContext, "Cluster Wait " + getInterfaceName());
    	}
    	
    	@Override
    	public void stop(RegisteredServiceContext serviceContext)
    			throws Exception {
    		if (this.clusterSendTpe != null) {
    			this.clusterSendTpe.shutdown();
    		}
    		
    		if (this.clusterWaitTpe != null) {
    			this.clusterWaitTpe.shutdown();
    		}
    		super.stop(serviceContext);
    	}
    	
    	@Override
    	public String toString() {
    		StringBuffer b = new StringBuffer();
    		b.append("STUB{")
    		 .append("name: ").append(getName())
    		 .append(", id: ").append(this.id)
    		 .append(", si: ").append(this.si)
    		 .append("}");
    		    		
    		return b.toString();
    	}
    	
    	/**
    	 * create client
    	 * 
    	 * @return
    	 */
    	protected DefaultTelnetClientAdaptor createTelnetClient() {
    		return getClientAdaptor(this.si, this);
    	}
    	
    	/**
    	 * 初始化
    	 * 
    	 */
    	public void init(RegisteredServiceContext serviceContext) {
    		DefaultTelnetClientAdaptor tca = createTelnetClient();    		    		
    		TelnetConnectorServiceAdaptor sa = new TelnetConnectorServiceAdaptor();    		
    		sa.setService(tca);
    		
    		sa.setTextProtocolDecoder(getTextProtocolDecoder()); // decoder    		
    		sa.setTextProtocolEncoder(getTextProtocolEncoder()); // encoder
    		sa.setDefaultStreamRecord(false);
    		
    		sa.init(); // INIT
    		
    		setServiceAdaptor(sa);
            setAddress(si.address);
            setClientNumberMax(1);        
            setEndpointExpression(EndpointGenerator.generate(
            		ClusterCommandSupport.ENDPOINT_ID).getExpression());       
            setInterfaceId("0");
            setInterfaceLogService(null);
            setInterfaceName("cluster");
            setIntervalTime(10000);
            setPort(si.port);       
            setStreamNameUtil(null);
            setStreamOutput(null);
            setThreadNumberTaskDispatcher(1);
            setThreadNumberTaskResultDispatcher(1);
            setThreadNumberExtendTaskResultProcessor(2);
            setThreadPoolCoreSize(2);
            setThreadPoolMaxSize(4);
            setThreadPoolKeepAliveTime(60);
            setTimeout(180000);
            setToEndpointExpression("-117");
            setWaitTaskMax(8090);
            setLogDir(null);
            setLogLevel(null);
            setStreamRecord(false);
            setStreamLogOutputService(null);
            setBundleContext(serviceContext.getBundleContext());
    	}
    }
}
